/*
 * Copyright Hundsun Technologies Inc. All Rights Reserved.
 */

package common

import (
	"fmt"
	"net/http"
	"sync"
	"time"

	"github.com/golang/protobuf/proto"
	"go.uber.org/atomic"
	"hundsun.com/hsl/hschain/common/keystore"
	"hundsun.com/hsl/hschain/common/log/logging"
	"hundsun.com/hsl/hschain/common/message"
	"hundsun.com/hsl/hschain/common/types"
	"hundsun.com/hsl/hschain/common/util/pb"
	consensustypes "hundsun.com/hsl/hschain/consensus/types"
	pbledger "hundsun.com/hsl/hschain/protos/ledger"
	"hundsun.com/hsl/hschain/protos/p2p"
	"hundsun.com/hsl/hschain/protos/seqcenter"
	prototypes "hundsun.com/hsl/hschain/protos/types"
	commstore "hundsun.com/hsl/hschain/store"
	"hundsun.com/hsl/hschain/store/mq"
)

// NewBlockSequenceServiceServer 新建序列化中心服务
func NewBlockSequenceServiceServer(conf *BlockSequenceServiceServerConfig) *BlockSequenceServiceServer {
	service := &BlockSequenceServiceServer{
		conf:       conf,
		wg:         &sync.WaitGroup{},
		running:    atomic.NewBool(false),
		blockQueue: make(chan *BlockAssign),
	}

	wClient, err := commstore.NewMqClient(mq.ModeWorker.String())
	if err != nil {
		panic(err)
	}
	wClient.Sub(types.SequenceCenterModule, mq.ModeWorker.String())
	service.workerClient = wClient
	tClient, err := commstore.NewMqClient(mq.ModeTopic.String())
	if err != nil {
		panic(err)
	}
	tClient.Sub(types.SequenceCenterModule, mq.ModeTopic.String())
	service.topicClient = tClient
	return service
}

// BlockSequenceServiceServerConfig 服务配置
type BlockSequenceServiceServerConfig struct {
	CheckBlockEnabled bool
	RequestQueueSize  int
	MaxBlockSize      uint32
}

// BlockSequenceServiceServer 实现protos/seqcenter/seqcenter.proto中定义的BlockSequenceService
type BlockSequenceServiceServer struct {
	baseclient           *BaseClient
	log                  logging.Log
	wg                   *sync.WaitGroup
	running              *atomic.Bool
	currentBlockSequence *atomic.Uint64
	conf                 *BlockSequenceServiceServerConfig
	sequenceSigner       keystore.Account
	workerClient         mq.Client
	topicClient          mq.Client
	blockQueue           chan *BlockAssign
}

// BlockAssign block to assign
type BlockAssign struct {
	Block        *pbledger.Block
	OriginPeerID string
}

// Start 启动
func (m *BlockSequenceServiceServer) Start(moduleLog logging.Log) error {
	m.log = moduleLog

	maxSequence, err := m.GetCurrentBlockSequence()
	if err != nil {
		return err
	}
	m.log.Infof("max BlockSequence of saved block: %d", maxSequence)

	m.currentBlockSequence = atomic.NewUint64(maxSequence)
	m.running.Store(true)
	m.EventLoop()

	return nil
}

// EventLoop 循环处理消息
func (m *BlockSequenceServiceServer) EventLoop() {
	go m.procTopicMqMsg()
	go m.procWorkerMqMsg()
	go m.procBlockQueue()
}

// GetBlockQueue get block queue
func (m *BlockSequenceServiceServer) GetBlockQueue() chan *BlockAssign {
	return m.blockQueue
}

// SetBaseClient set base client
func (m *BlockSequenceServiceServer) SetBaseClient(client *BaseClient) {
	m.baseclient = client
}

// Stop 停止
func (m *BlockSequenceServiceServer) Stop() {
	if m.wg != nil {
		m.wg.Wait()
	}
	m.running.Store(false)
}

// ProcessRelayMessage 处理中转消息
func (m *BlockSequenceServiceServer) ProcessRelayMessage(msg *mq.Message) {
	if !m.running.Load() {
		m.log.Error("Block sequense is not running.")
		return
	}

	p2pMsg := msg.Data.(*p2p.P2PMsg)
	relayMsg, err := message.DecodeMessage(p2pMsg)
	if err != nil {
		m.log.Errorf("Decode relay message failed, err:%s\n", err.Error())
		return
	}

	msgType := relayMsg.MsgType
	if msgType == consensustypes.BlockSequenceAddBlockMsg {
		m.AssignBlock(relayMsg.Payload, p2pMsg.PeerId)
	} else if msgType == consensustypes.BlockSequenceRequestMsg {
		m.ReplyCurrentSequence(msg, p2pMsg.PeerId)
	} else {
		m.log.Warnf("BlockSequence server receive unknown message: %s", msgType)
	}
}

// AssignBlock 区块放入处理队列
func (m *BlockSequenceServiceServer) AssignBlock(payload []byte, peerID string) {
	var block = &pbledger.Block{}
	err := proto.Unmarshal(payload, block)
	if err != nil {
		m.log.Error("Unmarshal relay msg failed, err:%s\n", err.Error())
		return
	}

	m.log.Infof("current block queue size: %d", len(m.blockQueue))
	m.log.Infof("Put block assign message into block queue, height: %d.", block.Header.Number)
	m.blockQueue <- &BlockAssign{
		Block:        block,
		OriginPeerID: peerID,
	}
}

// ReplyCurrentSequence 回复当前sequence
func (m *BlockSequenceServiceServer) ReplyCurrentSequence(msg *mq.Message, peerID string) {
	currentSequence := m.currentBlockSequence.Load()
	seq := &prototypes.Uint64{I: currentSequence}
	payload, err := proto.Marshal(seq)
	if err != nil {
		m.log.Errorf("sendSaveBlockResponse marshal resp failed, err:%s\n", err.Error())
		return
	}

	var relayMsg = &consensustypes.ConsensRelayMsg{}
	relayMsg.Payload = payload
	relayMsg.MsgType = consensustypes.BlockSequenceResponseMsg
	respMsg, err := message.EncodeMessage(peerID, relayMsg)
	if err != nil {
		m.log.Errorf("sendSaveBlockResponse encode p2p message failed, err:%s\n", err.Error())
		return
	}

	m.log.Debugf("Send consensus relay response, current sequence: %d.", currentSequence)
	replyMsg := m.workerClient.NewMessage(types.P2pModule, mq.ModeWorker.String(), types.EventConsensusResponseMsg, respMsg)
	msg.Reply(replyMsg)
}

// SaveBlock 保存区块
func (m *BlockSequenceServiceServer) SaveBlock(block *pbledger.Block) (*seqcenter.SaveBlockResponse, error) {
	err := m.preValidateBlock(block)
	if err != nil {
		return nil, err
	}

	resp, err := m.blockSequenceAssign(block)
	if err != nil {
		return nil, err
	}

	return resp, nil
}

// GetCurrentSequence 获取当前序列号
func (m *BlockSequenceServiceServer) GetCurrentSequence() uint64 {
	if m.currentBlockSequence != nil {
		return m.currentBlockSequence.Load()
	}
	return 0
}

func (m *BlockSequenceServiceServer) procBlockQueue() {
	for blockToAssign := range m.blockQueue {
		m.log.Infof("Begin to process block %d in block queue.", blockToAssign.Block.Header.Number)
		var resp = &seqcenter.SaveBlockResponse{}
		blk := blockToAssign.Block
		originPeerID := blockToAssign.OriginPeerID
		resp, err := m.SaveBlock(blk)
		if err != nil {
			m.log.Errorf("Save block failed, err: %s", err.Error())
			resp = m.createBadResponseAndSend(err, blk)
		}

		m.log.Info("Save block and send sequence add block response.")
		// 处理请求来自外部节点，构造p2p消息转发回去
		if originPeerID != "" {
			m.sendSaveBlockResponse(originPeerID, resp)
		}

		err = m.afterSaveBlockSequence(blk, resp)
		if err != nil {
			m.log.Errorf("After save block sequence failed, %s", err.Error())
			continue
		}
		m.log.Infof("After save block sequence process successfully.")

	}
}

func (m *BlockSequenceServiceServer) procTopicMqMsg() {
	for msg := range m.topicClient.Recv() {
		m.log.Infof("SequenceCenter topic client recv msg: %s", types.GetEventName(int(msg.Ty)))
		switch msg.Ty {
		case types.EventConsensusRequestMsg:
			mq.ProcessMsg(msg, m.ProcessRelayMessage, nil, m.log)
		}
	}
}

func (m *BlockSequenceServiceServer) procWorkerMqMsg() {
	for msg := range m.workerClient.Recv() {
		m.log.Infof("SequenceCenter worker client recv msg ty %s", types.GetEventName(int(msg.Ty)))
		switch msg.Ty {
		case types.EventConsensusSyncRequestMsg:
			mq.ProcessMsg(msg, m.ProcessRelayMessage, nil, m.log)
		}
	}
}

func (m *BlockSequenceServiceServer) preValidateBlock(block *pbledger.Block) error {
	var err error
	rawBlk := block
	reqLogger := pb.GetBlockLog(m.log, rawBlk)
	reqLogger.Debug("start validating...")
	blockBytes, _ := proto.Marshal(block)

	if len(blockBytes) > int(m.conf.MaxBlockSize) {
		return fmt.Errorf("blockSize '%d' exceed limit '%d'", len(blockBytes), m.conf.MaxBlockSize)
	}

	err = m.baseclient.ValidateBasicInfoOfBlock(rawBlk)
	if err != nil {
		return err
	}

	if !rawBlk.Verify() {
		return types.ErrInvalidSignature
	}
	reqLogger.Debug("block signature validation passed")
	return nil
}

// runBlockSequenceAssign 分配序列号并保存区块
func (m *BlockSequenceServiceServer) blockSequenceAssign(block *pbledger.Block) (*seqcenter.SaveBlockResponse, error) {
	nextSeq := m.currentBlockSequence.Load() + 1
	m.log.Infof("started... nextBlockSequence:%d", nextSeq)

	reqLogger := m.log.WithField(logging.LogFieldBlockNumber, block.Header.Number).WithField(logging.LogFieldBlockSequence, nextSeq)
	reqLogger.Debug("try to assign sequence")
	block.Header.BlockSequence = nextSeq
	resp := &seqcenter.SaveBlockResponse{
		StatusCode:    http.StatusOK,
		ErrorMessage:  "",
		BlockNumber:   block.Header.Number,
		BlockHash:     block.Hash(),
		BlockSequence: nextSeq,
		Signature:     nil,
	}
	resp.Sign(m.sequenceSigner.GetKeyPair())
	reqLogger.Info("sequence signed")
	block.Header.BlockSequenceSignature = resp.Signature
	err := m.SaveCurrentBlockSequence(nextSeq)

	if err != nil {
		return nil, err
	}
	reqLogger.Info("block saved")
	m.currentBlockSequence.Inc()
	m.log.Debugf("current block sequence: %d", m.currentBlockSequence.Load())

	return resp, nil
}

func (m *BlockSequenceServiceServer) sendSaveBlockResponse(originPeerID string, resp *seqcenter.SaveBlockResponse) {
	var relayMsg = &consensustypes.ConsensRelayMsg{}
	payload, err := proto.Marshal(resp)
	if err != nil {
		m.log.Errorf("sendSaveBlockResponse marshal resp failed, err:%s\n", err.Error())
		return
	}
	relayMsg.Payload = payload
	relayMsg.MsgType = consensustypes.BlockSequenceAddBlockRespMsg
	respMsg, err := message.EncodeMessage(originPeerID, relayMsg)
	if err != nil {
		m.log.Errorf("sendSaveBlockResponse encode p2p message failed, err:%s\n", err.Error())
		return
	}
	msg := m.topicClient.NewMessage(types.P2pModule, mq.ModeTopic.String(), types.EventConsensusRequestMsg, respMsg)
	err = m.topicClient.Send(msg, false)
	if err != nil {
		m.log.Errorf("sendSaveBlockResponse send response failed, err:%s\n", err.Error())
		return
	}
	m.log.Debugf("sendSaveBlockResponse send response successfully.")
}

func (m *BlockSequenceServiceServer) createBadResponseAndSend(err error, block *pbledger.Block) *seqcenter.SaveBlockResponse {
	response := &seqcenter.SaveBlockResponse{
		StatusCode:    http.StatusBadRequest,
		ErrorMessage:  err.Error(),
		BlockNumber:   block.Header.Number,
		BlockHash:     block.Hash(),
		BlockSequence: 0,
		Signature:     nil,
	}
	response.Sign(m.sequenceSigner.GetKeyPair())

	return response
}

func (m *BlockSequenceServiceServer) afterSaveBlockSequence(block *pbledger.Block, resp *seqcenter.SaveBlockResponse) error {
	mLog := m.log.WithField(logging.LogFieldBlockNumber, resp.BlockNumber)

	err := m.baseclient.WriteBlock(block)
	if err != nil {
		mLog.Errorf("dispatched new block to blockchain module failed, err:%s\n.", err.Error())
		return err
	}
	mLog.Debug("dispatched new block to blockchain module.")
	return m.baseclient.BroadCastBlock(block, 5)
}

// SaveCurrentBlockSequence 保存当前区块sequence
func (m *BlockSequenceServiceServer) SaveCurrentBlockSequence(blockSeq uint64) error {
	msg := m.topicClient.NewMessage(types.BlockChainModule, mq.ModeTopic.String(), types.EventStoreCurrentBlockSequence, blockSeq)
	err := m.topicClient.Send(msg, false)
	if err != nil {
		return err
	}

	return nil
}

// GetCurrentBlockSequence 获取当前最新sequence
func (m *BlockSequenceServiceServer) GetCurrentBlockSequence() (uint64, error) {
	msg := m.workerClient.NewMessage(types.BlockChainModule, mq.ModeWorker.String(), types.EventQueryCurrentBlockSequence, nil)
	err := m.workerClient.Send(msg, true)
	if err != nil {
		return 0, err
	}

	resp, err := m.workerClient.WaitTimeout(msg, time.Second*consensustypes.DefaultMqTimeOut)
	if err != nil {
		return 0, err
	}

	currentBlockSequence := resp.Data.(uint64)
	return currentBlockSequence, nil
}
